package com.abyss.avro;

import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Map;

/**
 * @Date 2020/9/12
 */
public class AvroSerializerSchema<T extends SpecificRecordBase> implements Serializer<T> {

    /**
     * 开发步骤：
     *  1.泛型参数继承avro基类，实现序列化接口
     *  2.重写序列化方法
     *  3.新建字节数组输出流对象
     *  4.获取二进制对象BinaryEncoder
     *  5.输出数据
     */
    @Override
    public void configure(Map configs, boolean isKey) {

    }

    //2.重写序列化方法
    @Override
    public byte[] serialize(String topic, T data) {

        //定义约束对象
        SpecificDatumWriter<T> writer = new SpecificDatumWriter<>(data.getSchema());

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        //二进制编码对象
        BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(bos,null);
        try {
            //将data数据写入二进制流
            writer.write(data,binaryEncoder);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return bos.toByteArray();
    }

    @Override
    public void close() {

    }
}